package com.example.bookservice.kafka;

import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.example.bookservice.entity.Book;
import com.example.bookservice.entity.Chapter;
import com.example.bookservice.service.BookService;
import com.example.bookservice.service.ChapterService;
import com.example.commom.enums.StatusCodeEnum;
import com.example.commom.exception.CustomException;
import com.example.commom.global.AppVariable;
import com.example.commom.global.KafkaConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;

/**
 *  kafka 订阅主题, 审核书籍和章节
 */
@Slf4j
@Configuration
@Transactional
public class KafkaCheckMessageReceiver {

    @Autowired
    private ChapterService chapterService;

    @Autowired
    private BookService bookService;

    /**
     * 审核书籍
     */
    @KafkaListener(topics = KafkaConstants.KAFKA_CHECK_BOOK_TOPIC, groupId = KafkaConstants.KAFKA_GROUP_ID)
    public void receiveCheckBookComment(String message, Acknowledgment acknowledgment) {

        try {
            log.info("收到消息: " + message);

            String[] strings = message.split(":");

            // 拿到 书籍 id 和 审核结果
            Long bookId = Long.parseLong(strings[0]);
            boolean status = Boolean.parseBoolean(strings[1]);

            // 修改书籍信息
            LambdaUpdateWrapper<Book> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(Book::getBookId, bookId)
                    .set(Book::getIsAvailable, status)
                    .set(Book::getIsApproved, true);

            boolean flag = bookService.update(updateWrapper);

            // 手动提交偏移量
            if (flag) {
                log.info(LocalDateTime.now() + ": id为: " + bookId +
                        " 的书籍: 审核完成!");
                acknowledgment.acknowledge();
            } else {
                throw new CustomException(StatusCodeEnum.FAILED.getMsg(), StatusCodeEnum.FAILED.getCode());
            }

        } catch (Exception e) {
            log.error(e.getMessage());
            throw new CustomException(StatusCodeEnum.FAILED.getMsg(), StatusCodeEnum.FAILED.getCode());
        }

    }


    /**
     * 审核章节
     */
    @KafkaListener(topics = KafkaConstants.KAFKA_CHECK_CHAPTER_TOPIC, groupId = "firstTopic001", containerFactory = "kafkaListenerContainerFactory")
    public void receiveCheckChapterComment(String message, Acknowledgment acknowledgment) {

        try {
            log.info("收到消息: " + message);

            String[] strings = message.split(":");

            // 拿到 章节id 和 审核结果
            Long chapterId = Long.parseLong(strings[0]);
            boolean status = Boolean.parseBoolean(strings[1]);

            // 修改书籍信息
            LambdaUpdateWrapper<Chapter> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(Chapter::getChapterId, chapterId)
                    .set(Chapter::getIsAvailable, status)
                    .set(Chapter::getIsApproved, true);

            boolean flag = chapterService.update(updateWrapper);

            // 手动提交偏移量
            if (flag) {
                log.info(LocalDateTime.now() + ": id为: " + chapterId +
                        " 的章节: 审核完成!");
                acknowledgment.acknowledge();
            } else {
                throw new CustomException(StatusCodeEnum.FAILED.getMsg(), StatusCodeEnum.FAILED.getCode());
            }
        } catch (Exception e) {
            log.error(e.getMessage());
            throw new CustomException(e.getMessage(), StatusCodeEnum.FAILED.getCode());
        }

    }
}
